package com.ecommerce.source;
import com.ecommerce.function.*;import com.ecommerce.model.*;import com.ecommerce.sink.ClickHouseSink;import com.ecommerce.source.ProductDataSource;import com.ecommerce.source.SearchWordDataSource;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.windowing.time.Time;
public class EcommerceMarketAnalysisJob {
    
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.enableCheckpointing(30000); // 30秒检查点
        
        // 1. 商品榜单处理
        processProductRanking(env);
        
        // 2. 搜索排行处理
        processSearchRanking(env);
        
        // 执行任务
        env.execute("电商市场分析平台");
    }
    
    private static void processProductRanking(StreamExecutionEnvironment env) {
        // 商品数据源
        DataStream<ProductData> productStream = env
            .addSource(new ProductDataSource())
            .name("product-source")
            .setParallelism(1);
        
        // 按交易总量排行
        DataStream<ProductData> totalTransactionRank = productStream
            .keyBy(ProductData::getCategory)
            .process(new ProductRankingFunction())
            .name("total-transaction-ranking")
            .setParallelism(4);
        
        // 按交易增速排行
        DataStream<ProductData> growthRateRank = productStream
            .keyBy(ProductData::getCategory)
            .process(new GrowthRateRankingFunction())
            .name("growth-rate-ranking")
            .setParallelism(4);
        
        // 写入ClickHouse
        totalTransactionRank.addSink(new ClickHouseSink<>("product_ranking", ProductData.class))
            .name("clickhouse-product-sink-total")
            .setParallelism(2);
            
        growthRateRank.addSink(new ClickHouseSink<>("product_ranking", ProductData.class))
            .name("clickhouse-product-sink-growth")
            .setParallelism(2);
    }
    
    private static void processSearchRanking(StreamExecutionEnvironment env) {
        // 搜索词数据源
        DataStream<SearchWordData> searchStream = env
            .addSource(new SearchWordDataSource())
            .name("search-word-source")
            .setParallelism(1);
        
        // 搜索人气排行
        DataStream<SearchWordData> searchPopularityRank = searchStream
            .keyBy(SearchWordData::getWordType)
            .process(new SearchRankingFunction("popularity"))
            .name("search-popularity-ranking")
            .setParallelism(4);
        
        // 搜索增速排行
        DataStream<SearchWordData> searchGrowthRank = searchStream
            .keyBy(SearchWordData::getWordType)
            .process(new SearchRankingFunction("growth"))
            .name("search-growth-ranking")
            .setParallelism(4);
        
        // 蓝海词挖掘
        DataStream<SearchWordData> blueOceanWords = searchStream
            .filter(word -> word.getSupplyDemandRatio() > 2.0)
            .keyBy(SearchWordData::getWordType)
            .process(new SearchRankingFunction("blue_ocean"))
            .name("blue-ocean-words")
            .setParallelism(4);
        
        // 写入ClickHouse
        searchPopularityRank.addSink(new ClickHouseSink<>("search_ranking", SearchWordData.class))
            .name("clickhouse-search-sink-popularity")
            .setParallelism(2);
            
        searchGrowthRank.addSink(new ClickHouseSink<>("search_ranking", SearchWordData.class))
            .name("clickhouse-search-sink-growth")
            .setParallelism(2);
            
        blueOceanWords.addSink(new ClickHouseSink<>("search_ranking", SearchWordData.class))
            .name("clickhouse-search-sink-blueocean")
            .setParallelism(2);
    }}